package com.king.app

import com.king.config.{DBConfig, StateBackendConfig}
import com.king.function.CustomerDeseriallization
import com.king.util.MyKafkaUtil
import com.ververica.cdc.connectors.mysql.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

/**
 * @Author: KingWang
 * @Date: 2022/1/15  
 * @Desc:
 **/
object FlinkCdcWithVerverica {
  def main(args: Array[String]): Unit = {

    //1. 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //1.1 开启ck并指定状态后端fs

    env.setStateBackend(new FsStateBackend(StateBackendConfig.getFileCheckPointDir("cdc_ververica")))
      env.enableCheckpointing(10000L) //头尾间隔：每5秒触发一次ck
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)  //
    env.getCheckpointConfig.setCheckpointTimeout(10000L)
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(10000l)  //尾和头间隔时间3秒

    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));



    //2. 通过flinkCDC构建SourceFunction并读取数据
    val dbServer = DBConfig.mysql_gmall_flink()
    val sourceFunction = MySqlSource.builder[String]()
      .hostname(dbServer.hostname)
      .port(dbServer.port)
      .username(dbServer.username)
      .password(dbServer.password)
      .databaseList("gmall-210325-flink")


      //如果不添加该参数，则消费指定数据库中所有表的数据
      //如果添加，则需要按照 数据库名.表名 的格式指定，多个表使用逗号隔开
//      .tableList("gmall-210325-flink.base_trademark")
//      .deserializer(new StringDebeziumDeserializationSchema())
      .deserializer(new CustomerDeseriallization())

      //监控的方式：
      // 1. initial 初始化全表拷贝，然后再比较
      // 2. earliest 最早的
      // 3. latest  指定最新的
      // 4. specificOffset 指定offset
      // 3. timestamp 比指定的时间大的

      .startupOptions(StartupOptions.initial())
      .build()

    val  dataStream = env.addSource(sourceFunction)

    //3. sink
    dataStream.print()
    val sinkTopic = "ods_base_db"
    dataStream.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic))

    //4. 启动任务
    env.execute("flink-cdc")

  }
}
